package sparkCore.net

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source
import scala.util.{Failure, Success}

/**
 * @Author:lixinlei
 * @Date:2021/5/19 8:40
 **/
object WashData {

  val sexFilePath = "E://data_resources/net/sexDictFile.csv"
  val netFilePath = "E://data_resources/net/netClean_test.csv"
  val resOutFilePath = "E://data_resources/net/washedData"

  /**
   * 把性别数据先封装成一个Map，方便后面使用
   * @return
   */
  def getSexMap():Map[String,String] = {

    val source = Source.fromFile(sexFilePath)

    val sexList = source.getLines().toList

    val sexMapBuff = Map[String,String]().toBuffer

    //对性别文件中的每行数据进行处理，读取的是字符串，拆分，生成键值对，添加到可变Map中
    sexList.foreach((line:String) => {
      val stuInfo = line.split(",")
      val stuId = stuInfo(0)
      val sex = stuInfo(1)
      sexMapBuff.append((stuId,sex))
    })

    source.close()

    sexMapBuff.toMap

  }

  /**
   * 给定清洗条件，过滤数据
   * a.每个字段不允许为空
   * b.结束时间必须大于开始时间
   * c.如果有表头需要处理
   * d.时间必须是0-23点
   * e.同一个人的上网时间不允许有交叉
   */
  def filterNetData(line:String):Boolean = {
    var boo = true
    //第一个判断条件，把表头过滤掉
    if(!line.contains("stuId") && !line.contains("startTime") && !line.contains("endTime")){
      val netLine = line.split(",")
      //如果最后一个字段为空，长度为2,但是正常执行，所有条件添加长度为3的限制
      if(netLine.size == 3){
        //如果有任何一个字段为空，boo都一定是false
        //scala中没有continue和break，但是可以通过其他方式实现。
        netLine.foreach((field:String)=>{
          if(null==field || field.equals("")){
            println("字段为空==="+line)
            boo = false
          }
        })
      }else{
        println("字段数量不是3==="+line)
        boo=false
      }

      //如果所有字段都不为空，boo为true，可以继续过滤第三个条件
      if(boo){
        val startTimeStr = netLine(1)
        val endTimeStr = netLine(2)

        //jdk1.8以后提供的LocalDateTime方法判断时间
//        val dtf = DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm")
//        val startTime_LD = LocalDateTime.parse(startTimeStr,dtf)
//        val endtTime_LD = LocalDateTime.parse(endTimeStr,dtf)

        val sdf = new SimpleDateFormat("yyyy/MM/dd HH:mm")

        //先把字符串类型的时间使用 Try类型捕获异常，处理异常。
        //如果时间正确，正常转换成Date类型数据
        //如果时间错误，直接把boo置为false
        //保证时间是符合自然规律的
        val startTime_try = util.Try(sdf.parse(startTimeStr))
        val startTime_date = startTime_try match {
          case Success(value) => value
          case Failure(exception) =>{
            println("开始时间转换失败==="+line)
            println("startTime==="+exception)
            boo = false
          }
        }

        val endTime_try = util.Try(sdf.parse(endTimeStr))
        val endTime_date = endTime_try match {
          case Success(value) => value
          case Failure(exception) => {
            println("结束时间转换失败==="+line)
            println("endTime==="+exception)
            boo = false
          }
        }

        //如果开始时间和返回时间都正确则两个变量都为Date类型
        //如果开始时间晚于结束时间，把boo置为false
        if (startTime_date.getClass == classOf[Date]
          && endTime_date.getClass ==classOf[Date]
          && startTime_date.asInstanceOf[Date].after(endTime_date.asInstanceOf[Date])) {
          println("开始时间晚于结束时间==="+line)
          boo = false
        }
      }
    }else{
      println("表头==="+line)
      boo = false
    }
    boo
  }

  /**
   * 清洗数据
   */
  def washNetData(sc:SparkContext) = {
    val sexMap = getSexMap()
    sc.textFile(netFilePath)
      .filter(filterNetData)
      .map((line:String) =>{
        val netInfo = line.split(",")
        val stuId = netInfo(0)
        val startTime = netInfo(1)
        val endTime = netInfo(2)
        val sex = sexMap(stuId)
        s"${stuId},${sex},${startTime},${endTime}"
    }).saveAsTextFile(resOutFilePath)
  }

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[1]").setAppName("netClean")
    val sc = new SparkContext(conf)
    washNetData(sc)
  }


}
